package com.dahantc;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

@Slf4j
public class TaskProducer extends Thread {

    private String topic;
    private int interval;

    public TaskProducer(String topic, int interval) {
        this.topic = topic;
        this.interval = interval;
    }

    @Override
    public void run() {
        setName("Thread-Producer-" + topic);
        log.info("--> TaskProducer [topic={},interval={}] is beginning...", topic, interval);
        Map<String, Object> configs = KafkaUtils.buildConfig(false);
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs);
        // producer.initTransactions();
        for (;;) {
            long wtime = System.currentTimeMillis();
            log.info("------------------------------------------------------------ [p-{}]", wtime%1000);
            log.info("--> Producer message from [{}] P-START.", topic);
            try {
                // producer.beginTransaction();
                // ⽤于封装Producer的消息
                ProducerRecord<String, String> record = new ProducerRecord<String, String>(
                        topic, // 主题名称
                        0, // 分区编号，现在只有⼀个分区，所以是0
                        System.currentTimeMillis() + "", // 数字作为key
                        "{\"appId\":\"emaApp001\",\"dstime\":1664399460000,\"id\":\"ff808081837e091a018385f046a01141\",\"status\":0}" // 字符串作为value
                );
                // 发送消息，同步等待消息的确认
                producer.send(record, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        if (exception == null) {
                            log.info("--> Topic [{}] producer successfully.", metadata);
                        } else {
                            log.error("--> Topic [{}] producer exception：" + exception.getMessage());
                        }
                    }
                });
                // producer.commitTransaction();
                log.info("--> Producer message from [{}] P-END.", topic);
                log.info("------------------------------------------------------------ [p-{}]\n", wtime%1000);
                sleep(interval * 1000);
            } catch (Exception e){
                e.printStackTrace();
                // producer.abortTransaction();
            }
        }
    }
}
